package com.aiwiown.snackmq.broker.handler.console;

import com.aiwiown.snackmq.broker.auth.AuthorizationService;
import com.aiwiown.snackmq.cluster.metadata.MetadataManager;
import com.aiwiown.snackmq.cluster.raft.SnackMQStateMachine;
import com.aiwiown.snackmq.common.dto.console.ClusterOverviewDTO;
import com.aiwiown.snackmq.common.handler.RequestContext;
import com.aiwiown.snackmq.common.message.Message;
import com.aiwiown.snackmq.common.message.MessageStatus;
import com.aiwiown.snackmq.common.message.MessageType;
import com.aiwiown.snackmq.common.serialization.JsonSerializer;
import com.aiwiown.snackmq.common.serialization.SerializerFactory;
import com.aiwiown.snackmq.management.BrokerMetrics;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class GetClusterOverviewHandler extends AbstractLeaderRequiredConsoleHandler {

    private final BrokerMetrics brokerMetrics;
    private final SnackMQStateMachine dataStateMachine;
    private final JsonSerializer serializer = SerializerFactory.getJsonSerializer();

    public GetClusterOverviewHandler(MetadataManager metadataManager, BrokerMetrics brokerMetrics, AuthorizationService authorizationService, SnackMQStateMachine dataStateMachine) {
        super(metadataManager, authorizationService);
        this.brokerMetrics = brokerMetrics;
        this.dataStateMachine = dataStateMachine;
    }

    @Override
    public MessageType getMessageType() {
        return MessageType.GET_CLUSTER_OVERVIEW_REQUEST;
    }

    @Override
    protected void doHandle(RequestContext ctx, Message request) {
        try {
            int brokerCount = metadataManager.getAllBrokers().size();
            int topicCount = metadataManager.getAllTopics().size();
            long totalMessages = dataStateMachine.calculateTotalClusterMessages();
            double tps = brokerMetrics.calculateClusterTps(totalMessages);
            ClusterOverviewDTO overview = new ClusterOverviewDTO(brokerCount, topicCount, totalMessages, tps);
            Message response = Message.builder()
                    .type(MessageType.RESPONSE)
                    .messageId(request.getMessageId())
                    .status(MessageStatus.SUCCESS)
                    .body(serializer.serialize(overview))
                    .build();
            ctx.writeResponse(response);
        } catch (Exception e) {
            log.error("Failed to get cluster overview", e);
            ctx.writeResponse(Message.builder()
                    .type(MessageType.RESPONSE)
                    .messageId(request.getMessageId())
                    .status(MessageStatus.FAILED)
                    .errorMessage("Failed to get cluster overview: " + e.getMessage())
                    .build());
        }
    }
}